package chapter4.eg8;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.junit.Test;

public class SpecificPartitionApp {
  @Test
  public void testSpecificPartition() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("group.id", "CountryCounter2");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "io.confluent.kafka.serializers.StringDeserializer");

    KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(properties);
    List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic");
    if (null != partitionInfos) {
      List<PartitionInfo> partitions = new ArrayList<>();
      for (PartitionInfo partitionInfo : partitionInfos) {
        partitions.add(partitionInfo);
      }

      while (true) {
        ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, GenericRecord> record : records) {
          System.out.printf("topic: %s, partition: %s, offset: %d, customer: %s, country: $s\n",
              record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }

        consumer.commitSync();
      }
    }
  }

}
